%%%-------------------------------------------------------------------
%%% @author wukai
%%% @copyright (C) 2020, <COMPANY>
%%% @doc
%%%
%%% @end
%%% Created : 29. 8月 2020 9:34 上午
%%%-------------------------------------------------------------------
-module(himq_mqtt).
-author("wukai").
-behaviour(ranch_protocol).
-include("himq.hrl").
-define(MQTT, mqtt).

%% API
-export([start_server/1]).
-export([start_link/3, init/3, loop/1, send/2, received/2, hibernate/1]).

start_server(_Opts) ->
  himq_mqtt_router:init_router(100),
  {ok, Ip} = inet:parse_address(himq_eenv:get(?HIMQ, ?LISTENER_MQTT_IP, "0.0.0.0")),
  Port = himq_eenv:get(?HIMQ, ?LISTENER_MQTT_PORT, 18830),
  SSL = himq_eenv:get(?HIMQ, ?LISTENER_MQTT_SSL, false),
  KeyFile = himq_eenv:get(?HIMQ, ?LISTENER_MQTT_SSL_KEY_FILE, undefined),
  CaFile = himq_eenv:get(?HIMQ, ?LISTENER_MQTT_SSL_CA_FILE, undefined),
  CaCertFile = himq_eenv:get(?HIMQ, ?LISTENER_MQTT_SSL_CAC_FILE, undefined),
  start_listener(SSL, {Ip, Port, CaFile, CaCertFile, KeyFile}).

start_listener(false, {Ip, Port, _CA, _Cert, _Key}) ->
  MaxConn = himq_eenv:get(?HIMQ, ?LISTENER_MQTT_MAX_CONN, 102400),
  MaxAcceptorNumber = himq_eenv:get(?HIMQ, ?LISTENER_MQTT_MAX_ACCEPTORS, 10),
  {ok, _} = ranch:start_listener(himq_server,
    ranch_tcp,
    #{socket_opts=>[{port, Port}, {ip, Ip}],
      num_acceptors=>MaxAcceptorNumber,
      max_connections=>MaxConn
    }, ?MODULE, []);

start_listener(true, {Ip, Port, CA, Cert, Key}) ->
  MaxConn = himq_eenv:get(?HIMQ, ?LISTENER_MQTT_MAX_CONN, 102400),
  MaxAcceptorNumber = himq_eenv:get(?HIMQ, ?LISTENER_MQTT_MAX_ACCEPTORS, 10),
  Opts = #{socket_opts => [{port, Port}, {ip, Ip}, {cacertfile, CA},
    {certfile, Cert}, {keyfile, Key}],
    num_acceptors=>MaxAcceptorNumber,
    max_connections=>MaxConn
  },

  {ok, _} = ranch:start_listener(okmq_server, ranch_ssl, Opts, ?MODULE, []).

start_link(Ref, Transport, Opts) ->
  Pid = proc_lib:spawn_link(?MODULE, init, [Ref, Transport, Opts]),
  {ok, Pid}.

init(Ref, Transport, _Opts = []) ->
  {ok, Socket} = ranch:handshake(Ref),
  process_flag(trap_exit, true),
  Transport:setopts(Socket, [{active, true}, {nodelay, true}]),
  PState = himq_mqtt_codec:p_init(),
  loop(#himq_conn{trans = Transport, socket = Socket, protocol = himq_mqtt_codec, p_state = PState}).

loop(State) ->
  try
    receive
      {tcp, _, Bin} ->
        loop(received(State, Bin));
      {tcp_closed, Reason} ->
        %%lager:error("tcp_closed ~p", [Reason]),
        closed(State, tcp_closed);
      {tcp_error, _, Reason} ->
        %%lager:error("tcp_error ~p", [Reason]),
        closed(State, tcp_error);
      {ssl, _, Bin} ->
        loop(received(State, Bin));
      {ssl_closed, _, _} ->
        closed(State, ssl_closed);
      {'EXIT', _From, {shutdown, reconn}} ->
        lager:info("client ~p reconnect... ", [State#himq_conn.p_state]);
      {'EXIT', _From, {shutdown, Reason}} ->
        closed(State, Reason);
      {'EXIT', _From, Reason} ->
        %%lager:info("other exit msg:~p", [Reason]),
        closed(State, Reason);
      {remove, Reason} ->
        %%lager:info("other exit msg:~p", [Reason]),
        closed(State, Reason);
      Msg ->
        loop(message(Msg, State))
    after State#himq_conn.receive_timeout ->
      timeout(State)
    end
  catch
    exit:ExitReason ->
      lager:info("exit ~p", [ExitReason]),
      closed(State, exit);
    error:ErrorReason ->
      lager:info("error ~p stackinfo:~p", [ErrorReason,erlang:get_stacktrace()]),
      closed(State, error)
  end.

received(State, Bin) ->
  NBuf = <<(State#himq_conn.buffer)/binary, Bin/binary>>,
  Decode = State#himq_conn.protocol,
  Decode:decode(State#himq_conn{buffer = NBuf}).

message(Msg, State) ->
  Decode = State#himq_conn.protocol,
  Decode:message(Msg, State).

%%进程休眠
hibernate(State) when State#himq_conn.type == ws ->
  self() ! {hibernate, State};
hibernate(State) when State#himq_conn.type == tcp ->
  proc_lib:hibernate(?MODULE, loop, [State]).

%%超时关闭
timeout(State) when State#himq_conn.client_id =:= undefined ->
  lager:error("tcp timeout auth ==false");

timeout(State) ->
  lager:error("link id ~p timeout",[State#himq_conn.client_id]),
  himq_session:del(State#himq_conn.client_id, "Timeout"),
  Decode = State#himq_conn.protocol,
  Decode:timeout(State).

closed(State, Reason) when State#himq_conn.client_id =:= undefined ->
  lager:error("tcp closed auth ==false,reason:~p",[Reason]);

closed(State, Reason) ->
  himq_session:del(State#himq_conn.client_id, Reason),
  Decode = State#himq_conn.protocol,
  Decode:closed(State).

send(State, Bin) when State#himq_conn.type == ws ->
  self() ! {send, {binary, Bin}},
  ok;

send(State, Bin) when State#himq_conn.type == tcp ->
  T = State#himq_conn.trans,
  S = State#himq_conn.socket,
  case T:send(S, Bin) of
    ok -> ok;
    _ ->
      lager:info("send time out ")
  end.